package com.wjq.demo.hystrix;

import com.google.common.collect.Lists;
import org.junit.Test;
import rx.Observable;
import rx.Observer;
import rx.Subscriber;
import rx.Subscription;
import rx.functions.Func0;
import rx.schedulers.Schedulers;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

/**
 * @author wjq
 * @since 2023-04-20
 */
public class RxJavaTest {

    @Test
    public void getUuid() {

        System.out.println(UUID.randomUUID().toString().replace("-", ""));
        System.out.println(UUID.randomUUID().toString().replace("-", ""));

    }

    public static void main(String[] args) {

        Observable.just(1, 2, 3)
                .map((integer) -> integer + 1)
                .subscribe((integer) -> {
                    System.out.println(integer);
                });
    }


    @Test
    public void testUnsubscribe() {
        Integer[] from = {0, 1, 2, 3, 4, 5, 6, 7};

        Observable.from(from).subscribe(new Subscriber<Integer>() {
            @Override
            public void onNext(Integer integer) {
                if (integer > 4) {
                    unsubscribe();
                }
                System.out.println(integer);
            }

            @Override
            public void onCompleted() {
            }

            @Override
            public void onError(Throwable e) {
                e.printStackTrace();
            }
        });
    }

    public static Observable<Integer> range(Integer start, int count) {

        return Observable.unsafeCreate(onSubscribe -> {
            for (int i = 0; i < count; i++) {
                onSubscribe.onNext(start + i);
            }
            onSubscribe.onCompleted();
        });
    }

    @Test
    public void testObservableCreate() {
        Observable<Object> objectObservable = Observable.unsafeCreate(Observer::onCompleted);
        objectObservable.subscribe(o -> {
                },
                e -> {
                },
                () -> System.out.println("finish"));


        Observable.empty().subscribe(o -> {
                },
                e -> {
                },
                () -> System.out.println("finish"));


        objectObservable = Observable.unsafeCreate((onSubscriber) -> {
        });

        objectObservable.subscribe(o -> {
                },
                e -> {
                },
                () -> System.out.println("finish"));


        Observable.never().subscribe(o -> {
                },
                e -> {
                },
                () -> System.out.println("finish"));

        System.out.println("finish");

        range(1, 10).subscribe(System.out::println);

    }


    public static void log(String s) {
        System.out.println(Thread.currentThread().getName() + ":" + s);
    }

    @Test
    public void testCache() {
        Observable<Object> observable = Observable.unsafeCreate(subscriber -> {
            log("Create");
            subscriber.onNext(42);
            subscriber.onCompleted();
        });
        log("starting");
        observable.subscribe(i -> log("Element A:" + i));
        observable.subscribe(i -> log("Element B:" + i));
        log("Exit");

        Observable<Object> cache = observable.cache();
        log("cache starting");
        cache.subscribe(i -> log("Element A:" + i));
        cache.subscribe(i -> log("Element B:" + i));
        log("cache exit");
    }

    @Test
    public void testDoOnNext() {
        Observable.just(1, 2, 3)
                .doOnNext(System.out::println)
                .subscribe(integer -> {
                });
    }

    @Test
    public void testFlatMap() {
        Observable.just(Observable.just(1), Observable.just(2), Observable.just(3))
                .flatMap(integerObservable -> integerObservable)
                .subscribe(integer -> System.out.println(integer));
    }


    @Test
    public void testDoOn() {
        Observable.just(1)
                .doOnCompleted(() -> log("complete1"))
                .doOnCompleted(() -> log("complete2"))
                .subscribe(num -> log("subscribe:" + num));


        Observable.unsafeCreate(subscriber -> {
                    log("Create");
                    subscriber.onNext(42);
                    subscriber.onCompleted();
                })
                .doOnCompleted(() -> log("complete"))
                .doOnSubscribe(() -> log("subscribe"))
                .doOnTerminate(() -> log("terminate"))
                .subscribe((integer -> log("subscribe：" + integer)));


        log("===========");

        Observable.defer(() -> {
                    log("Create");
                    return Observable.just(42);
                })
                .doOnCompleted(() -> log("complete"))
                .doOnSubscribe(() -> log("subscribe"))
                .doOnTerminate(() -> log("terminate"))
                .subscribe((integer -> log("subscribe：" + integer)));
    }

    @Test
    public void testDoOnSubscribe() {
        Observable.defer(new Func0<Observable<Integer>>() {
                    @Override
                    public Observable<Integer> call() {
                        try {
                            log("first?");
                            return Observable.just(1);
                        } catch (Throwable ex) {
                            return Observable.error(ex);
                        }
                    }
                }).doOnSubscribe(() -> log("subscribe"))
                .subscribe((integer) -> log("accept:" + integer));

    }

    private static String dataProducer() {
        String str = "Hello,RxJava!";
        System.out.println("Produce Data :" + str);
        return str;
    }

    @Test
    public void testDefer() {

        log("just===================");
        Observable.just(dataProducer()).doOnSubscribe(() -> log("Subscribe!"))
                .subscribe(s -> log("Consume Data :" + s));

        log("defer===================");
        Observable.defer(() -> Observable.just(dataProducer()))
                .doOnSubscribe(() -> log("Subscribe!"))
                .subscribe(s -> log("Consume Data :" + s));

        log("complex defer===================");
        Observable.defer(() -> {
                    log("here1");
                    return Observable.defer(() -> {
                        log("here2");
                        return Observable.just(dataProducer());
                    }).map(s -> s.concat("!!"));
                })
                .doOnSubscribe(() -> log("Subscribe!"))
                .subscribe(s -> log("Consume Data :" + s));
    }


    @Test
    public void testLift() {

        Observable.just(1, 2)
                .lift(child -> new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        child.onCompleted();
                    }

                    @Override
                    public void onError(Throwable e) {
                        child.onError(e);
                    }

                    @Override
                    public void onNext(Integer o) {
                        if (o == 1) {
                            child.onNext(o);
                        }
                    }
                })
                .subscribe(integer -> log("finish:" + integer));
    }


    @Test
    public void testWindow() {
        log("current time:" + LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME));
        Subscription subscribe = Observable.interval(100, TimeUnit.MILLISECONDS)
                .window(500, TimeUnit.MILLISECONDS)
                .flatMap(Observable::toList)
                .startWith(Lists.newArrayList(1L, 2L, 3L), Lists.newArrayList(1L, 2L, 3L), Lists.newArrayList(1L, 2L, 3L), Lists.newArrayList(1L, 2L, 3L), Lists.newArrayList(1L, 2L, 3L))
                .window(5, 1)
                .flatMap(Observable::toList)
                .subscribe(l -> log("current time:" + LocalDateTime.now().format(DateTimeFormatter.ISO_DATE_TIME) + ", result:" + l));

        try {
            TimeUnit.SECONDS.sleep(40);
        } catch (InterruptedException e) {

        }
        subscribe.unsubscribe();
    }


    @Test
    public void testSubscribeOn() {

        Observable.unsafeCreate(subscriber -> {
                    log("start");
                    subscriber.onNext(1);
                    subscriber.onNext(2);
                    log("complete");
                    subscriber.onCompleted();

                })
                .subscribeOn(Schedulers.newThread())
                .subscribe(integer -> log(String.valueOf(integer)));
    }
}
